package com.foton.lotteryapplication.mq;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

/**
 * @Author RenPu
 * @Date 2024/1/8 11:28
 * @Version 1.0
 * @Description: kafka 消息生产者 -Test
 **/

@Slf4j
@Component
public class KafkaProducerTest {


    @Resource
    private KafkaTemplate<String,Object> kafkaTemplate;

    // 定义主题以及消费组
    public final static String TOPIC_TEST="Hello-Kafka";
    public final static String TOPIC_GROUP="test-consumer-group";

    // 产生消息并发送到主题,通过回调函数检测发送消息的结果

    public void send(Object obi){
        String msg = JSONUtil.toJsonStr(obi);
        log.info("准备发送的消息为：：：{}",msg);

        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_TEST, msg);

        //通过回调函数检测发送消息的结果
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable throwable) {
                log.error(TOPIC_TEST+"生产者 == 消息发送失败 ==Failure==具体异常如下 ；；；{}",throwable.getLocalizedMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                log.info(TOPIC_TEST+"生产者 == 消息发送成功 == SUCCESS == ；；；{}",stringObjectSendResult.toString());

            }
        });



    }




}
